#include "types/bptree.h"
#undef debug
#define debug 

UINT32 NODE_TYPE_MASK = 0x01; //Will and to header to get the node type

enum node_type
{
  inter_node,
  leaf_node
};

bool bptree::index_debug = false;

bptree::bptree(string name) : bm (BufferManager::inst())
{
  //printf("\nIn bptree constructor %s",name.c_str());
  read_block(name,0,(BYTE *)&header);
  indexName = name;
  cacheId = 0;
}

bptree::bptree(string name, bool dupAllowed, vector<index_data> keys) : bm (BufferManager::inst())
 
{
	// BufferManager &bm = BufferManager::inst();
	// //Create an empty index
	// bm.create_table(name);
	// index_header header;
	// header.dupAllowed = (dupAllowed) ? 1 : 0;
	// write_block(name, 0, (BYTE *)&header);
// =======
//  bm = BufferManager::inst();
  /* Create a File */
  bm.create_index(name);

  indexName = name;

  // Update the Header Initially
  header.dupAllowed = dupAllowed;
  header.numLevels = 0;
  header.numBlocks = 1;
  memset(header.reserved,0x00,42);

  //write to file
  write_block(name, 0, (BYTE *)&header);

  //Create the first root node
  struct data_entry Node;

  memset(&Node,0x00,BLKSIZE);
  Node.LeafNode = 1;

  write_block(name, 1, (BYTE *)&Node);

  cacheId = 0;
  
  bool createComplete = false;
  if(keys.size() != 0)
  {
	createComplete = bulk_insert(keys);
   if(!createComplete)
	{
	  bm.delete_index(name);
	  throw Exception(recmgr_dup_attr);
	}
 }

}

inline bool compare_index_data(index_data d1, index_data d2)
{
	return (d1.key < d2.key);
}

bool bptree::bulk_insert(vector<index_data> keys) 
{
	// vector<> index_blocks;
	// int num_iblocks = 0;
	// index_data cur_data, prev_data;
	// //sort the keys
	// sort(keys.begin(), keys.end(), compare_index_data);
	// //Create data blocks on disk
	// //Index blocks on memory
	// for(int i=0; i<keys.size(); i++)
	// {
		// for(int j=0; j<NUM_LEAF_DATA; j++)
		// {
			
		// }
	// }
// =======
	for(unsigned int i=0; i<keys.size();i++)
	{
//		printf(" Data being inserted: %d \n",keys[i].key);
		if(RecInsert(keys[i]))
			return false;
	}
	return true;
}

bool bptree::update(index_data data) 
{
	struct data_entry *Node;
	struct index_entry *pNode;

	UINT32 Key = data.key;
	UINT32 Ptr = 0;

	UINT32 KIdx, Split = 0;
	int EqualKey = 0; 

	UINT32 blockNum = 0;

	pNode = (struct index_entry *)&Node;

	//Assumption Lookup is done before update is called
	//Lookup and get the correct Node
	if(!cacheId)
		return 0;
	Lookup(data.key,pNode,0);
	Node = (struct data_entry *)&cache[--cacheId].cacheBlock;
	blockNum = cache[cacheId].blockNum;

	blockNum = cache[cacheId-1].blockNum;

	KIdx = FindKey(Node, Key);

	//Have got the correct Node where it has to be inserted
	EqualKey = KIdx && Node->data[KIdx-1].key == Key;

	if(EqualKey)
	{
		if(header.dupAllowed)
		{
			/* (KIdx - 1) is the location where the Key has been found */
			HandleDup(Node,data,KIdx-1,blockNum);
			return false;
		}
		else
		{
			debug(" No duplicates allowed : %d \n",Key);
			return true;
		}
	}

	/*New Key insert normally */
	do
	{

		if(Node->LeafNode)
		{
			Split = InsertKeyOnLeaf(Node, KIdx, data, &Key, &Ptr, &blockNum); 
			write_block(indexName,blockNum, (BYTE *)Node);
		}
		else
		{
			Split = InsertKeyOnNonLeaf(pNode, KIdx, &Key, &Ptr,&blockNum); 
			write_block(indexName,blockNum,(BYTE *)pNode);
		}

		//Load the parent
		if(cacheId >0 && Split)
		{
			pNode = (struct index_entry *)&cache[--cacheId].cacheBlock;

			blockNum = cache[cacheId].blockNum;

			KIdx = FindKey((struct data_entry*)pNode, Key);
		}

	}while(Split);

	cacheId = 0;
	return 0; 
}

vector<index_data> bptree::lookup(Condition cond) 
{
	vector<index_data> result;

	if(cond.type == val_check)
	{
		//Check the condition, depending on that call the appropriate function
		switch(cond.op)
		{
			//Assuming lvalue is given for one Search operations with one Key
			case eq: result = search(cond.rvalue); break;
			case gt: result = searchGreaterThan(cond.rvalue); break;
			case lt: result = searchLessThan(cond.rvalue); break;
			case ge: result = searchGreaterThan(cond.rvalue - 1); break;
			case le: result = searchLessThan(cond.rvalue + 1); break;
			case ne:
			{
					 vector<index_data> vec1,vec2;
					 vec1 = searchLessThan(cond.lvalue);
					 vec2 = searchGreaterThan(cond.lvalue);
		//			 result(vec1.size() + vec2.size());
		//			 vector<index_data>::iterator iter =
		//				 set_union(vec1.begin(), vec1.end(), vec2.begin(),vec2.end(),result.begin());


 //Need to merge the two vectors and send the result back
//result = searchLessThan(cond.lvalue) + searchGreaterThan(cond.lvalue); 
//Merging TO Do
					 result = vec1;
					 }
					 break;
 /*If a new case is added to search between two Key values */
 /*Use this function: searchRange(cond.lvalue,cond.rvalue);*/
			default: debug("Bad Operation \n"); break;
		}

		return result;
	}
}

vector<index_data> bptree::lookup(UINT32 key) 
{
	vector<index_data> result;
	int KIdx,EqualKey = 0;
	struct index_entry Node;
	struct data_entry *pNode = (struct data_entry *)&Node;

	//Do a lookup

	Lookup(key,&Node,0);

	KIdx = FindKey(pNode, key);

	//Have got the correct Node
	EqualKey = KIdx && pNode->data[KIdx-1].key == key;

	if(EqualKey)
		result.push_back(pNode->data[KIdx-1]);

	return result;
}

void bptree::Lookup(long Key,struct index_entry *Node,UINT32 lookUpFlag)
{
	int index = 1;
	for(;;) 
	{
		read_block(indexName,index, (BYTE *)Node);

		if(!lookUpFlag)
		{
			//Copy into Cache Block
			memcpy(&cache[cacheId++].cacheBlock[0],Node,BLKSIZE);
			cache[cacheId-1].blockNum = index;
		}
		else
		{
			if(index_debug == true)
			{
				/*Printing the index buckets that are read for a select command */
				printf("\nBlock  %d: \t",index);

				if(index == 1)
					printf("root \n\t\t");
				else if(!Node->LeafNode)
					printf("Internal Node \n");

				bpPrintNode(Node);
			}
		}

		index = FindKey((struct data_entry *)Node, Key); 
		if(Node->LeafNode) 
			break; 

		index = Node->child[index]; 
	} 
	return ; 
} 

int  bptree::checkIfOvwFlowExists(struct data_entry *Node,UINT32 key)
{
	int ovwFlow = 0;
	if(Node->ovwFlowMask)
	{
		switch(key)
		{
			case 0: if(Node->ovwFlowMask & MASK_FIRST_KEY_OVERFLOW)		ovwFlow = 1; break;
			case 1: if(Node->ovwFlowMask & MASK_SECOND_KEY_OVERFLOW)	ovwFlow = 1; break;
			case 2: if(Node->ovwFlowMask & MASK_THIRD_KEY_OVERFLOW)		ovwFlow = 1; break;
			case 3: if(Node->ovwFlowMask & MASK_FOURTH_KEY_OVERFLOW)	ovwFlow = 1; break;
		}
	}
	return ovwFlow;
}

vector<index_data> bptree :: searchLessThan(UINT32 key)
{
	struct data_entry Node;
	struct index_entry *pNode;
	UINT32 KIdx;
	UINT32 EqualKey,ovwFlow=0;
	vector<index_data> result;
	int j;

//	printf("Select less than : Data < %d\n",key);

	if(key == 0)
	{
		debug("Key cannot be Zero \n");
		return result;
	}

	pNode = (struct index_entry *)&Node;

	Lookup(key,pNode,1);

	//Search the KeyIndex in the given Node
	KIdx = FindKey(&Node, key);

	EqualKey = KIdx && Node.data[KIdx-1].key == key;
	if(EqualKey)
	{
		KIdx-=2;
	}
	else
	{
		KIdx--;
	}

	/*Outer Loop for the entire range */
	while(1)
	{
		/*Inner Loop within the Block */
		for (j=KIdx; j>=0; j--)
		{
			/*Innermost Loop to print the OverFlow Node */
			if(header.dupAllowed)
			{
				/*Check if duplicate is present */
				if(checkIfOvwFlowExists(&Node,j))
				{
					/* Print from the OverFlow Node */
					struct data_entry ovwFlowNode;
					int index = Node.data[j].id.block_num;

					do
					{
						/*Read theS Overflow Node */
						read_block(indexName,index,(BYTE *)&ovwFlowNode);

						for(int i=0;i<ovwFlowNode.keyCount;i++)
						{
							result.push_back(ovwFlowNode.data[i]);

							debug("KEY: %d Ptr: %d %d\n",
									ovwFlowNode.data[i].key,ovwFlowNode.data[i].id.block_num,
									ovwFlowNode.data[i].id.slot_num);
						}
						index = ovwFlowNode.right;

					}while(index);
				}
				else
				{
					result.push_back(Node.data[j]);

					debug("KEY: %d Ptr: %d %d\n",Node.data[j].key,
							Node.data[j].id.block_num,Node.data[j].id.slot_num);
				}
			}
			else
			{
				result.push_back(Node.data[j]);
				debug("KEY: %d Ptr: %d %d \n",Node.data[j].key,
						Node.data[j].id.block_num,Node.data[j].id.slot_num);
			}
		}
		/*Move to the Left Block */
		/*If it exists */
		if(j<0 && Node.left)
		{
			read_block(indexName,Node.left,(BYTE *)&Node);
			KIdx=Node.keyCount - 1;
		}
		else
			break;
	}
	return result;
}

void bptree::print(string what)
{
	bpPrint();
}

int bptree::FindKey(struct data_entry *Node, UINT32 Key) 
{ 
	UINT32 k; 
	struct index_entry *idNode;
	if(!Node->LeafNode)
	{
		idNode = (struct index_entry *)Node;
		for(k = 0; k < idNode->keyCount; k++)
			if(idNode->key[k] > Key) 
				break; 
		return k; 
	}
	/*For leaf node*/
	for(k = 0; k < Node->keyCount; k++)
		if(Node->data[k].key > Key) 
			break; 
	return k; 
}

vector<index_data> bptree :: searchGreaterThan(UINT32 key)
{
	struct data_entry Node;
	UINT32 KIdx;
	UINT32 ovwFlow=0;
	vector<index_data> result;
	UINT32 j;

	if(key == 0)
	{
		debug("Key cannot be Zero \n");
		return result;
	}

	Lookup(key,(struct index_entry *)&Node,1);

	//Search the KeyIndex in the given Node
	KIdx = FindKey(&Node, key);

	/*Outer Loop for the entire range */
	while(1)
	{
		/*Inner Loop within the Block */
		for (j=KIdx; j<Node.keyCount; j++)
		{
			/*Innermost Loop to print the OverFlow Node */
			ovwFlow=0;

			if(header.dupAllowed)
			{
				/*Check if duplicate is present */
				if(checkIfOvwFlowExists(&Node,j))
				{
					/* Print from the OverFlow Node */
					struct data_entry ovwFlowNode;
					int index = Node.data[j].id.block_num;

					do
					{
						/*Read the Overflow Node */
						read_block(indexName,index,(BYTE *)&ovwFlowNode);

						for(int i=0;i<ovwFlowNode.keyCount;i++)
						{
							result.push_back(ovwFlowNode.data[i]);

							debug("KEY: %d Ptr: %d %d\n",
									ovwFlowNode.data[i].key,ovwFlowNode.data[i].id.block_num,
									ovwFlowNode.data[i].id.slot_num);
						}
						index = ovwFlowNode.right;

					}while(index);
				}
				else
				{
					result.push_back(Node.data[j]);

					debug("KEY: %d Ptr: %d %d\n",Node.data[j].key,
							Node.data[j].id.block_num,Node.data[j].id.slot_num);
				}
			}
			else
			{
				result.push_back(Node.data[j]);

				debug("KEY: %d Ptr: %d %d\n",Node.data[j].key,
						Node.data[j].id.block_num,Node.data[j].id.slot_num);
			}
		}
		/*Move to the Left Block */
		/*If it exists */
		if(j==Node.keyCount && Node.right)
		{
			read_block(indexName,Node.right,(BYTE *)&Node);
			KIdx=0;
		}
		else
			break;
	}
	return result;
}

vector<index_data> bptree :: searchRange(UINT32 key1,UINT32 key2)
{
	struct data_entry Node;
	UINT32 KIdx,j;
	UINT32 EqualKey;
	vector<index_data> result;

	if(key1 == 0 || key2 == 0)
	{
		debug("For range, keys should not be zero \n");
		return result;
	}

	Lookup(key1,(struct index_entry *)&Node,1);

	//Search the KeyIndex in the given Node
	KIdx = FindKey(&Node, key1);

	EqualKey = KIdx && Node.data[KIdx-1].key == key1;

	if(EqualKey)
	{
		KIdx--;
	}
	/*Outer Loop for the entire range */
	while(1)
	{
		/*Inner Loop within the Block */
		for (j=KIdx; j<Node.keyCount && Node.data[j].key <key2; j++)
		{
			/*Innermost Loop to print the OverFlow Node */
			if(header.dupAllowed)
			{
				/*Check if duplicate is present */
				if(checkIfOvwFlowExists(&Node,j))
				{
					/* Print from the OverFlow Node */
					struct data_entry ovwFlowNode;
					int index = Node.data[j+1].id.block_num;

					do
					{
						/*Read the Overflow Node */
						read_block(indexName,index,(BYTE *)&ovwFlowNode);

						for(int i=0;i<ovwFlowNode.keyCount;i++)
						{
							result.push_back(ovwFlowNode.data[i]);
							debug("KEY: %d Ptr: %d %d\n",
									ovwFlowNode.data[i].key,ovwFlowNode.data[i].id.block_num,
									ovwFlowNode.data[i].id.slot_num);
						}
						index = ovwFlowNode.right;

					}while(index);
				}
				else
				{
					result.push_back(Node.data[j]);

					debug("KEY: %d Ptr: %d %d\n",Node.data[j].key,
							Node.data[j].id.block_num,Node.data[j].id.slot_num);
				}
			}
			else
			{
				result.push_back(Node.data[j]);

				debug("KEY: %d Ptr: %d %d\n",Node.data[j].key,
						Node.data[j].id.block_num,Node.data[j].id.slot_num);
			}
		}
		/*Move to the next Block */
		/*If it exists */
		if(j==Node.keyCount && Node.right)
		{
			read_block(indexName,Node.right,(BYTE *)&Node);
			KIdx=0;
		}
		else
			break;
	}
	return result;
}

vector<index_data> bptree :: search(UINT32 key)
{
	struct data_entry Node;
	struct index_entry *pNode;
	UINT32 KIdx;
	UINT32 EqualKey;
	vector<index_data> result;

	pNode = (struct index_entry *)&Node;
	//First Lookup for the Key
	Lookup(key,pNode,1);

	//Search the KeyIndex in the given Node
	KIdx = FindKey(&Node, key);

	//Check if it exists
	//Have got the correct Node where it has to be inserted
	EqualKey = KIdx && Node.data[KIdx-1].key == key;

	if(EqualKey)
	{
		if(header.dupAllowed)
		{
			//*Check if duplicate is present */
			if(checkIfOvwFlowExists(&Node,KIdx-1))
			{
				/* Print from the OverFlow Node */
				struct data_entry ovwFlowNode;
				int index = Node.data[KIdx-1].id.block_num;

				do
				{
					/*Read the Overflow Node */
					read_block(indexName,index,(BYTE *)&ovwFlowNode);

					for(int i=0;i<ovwFlowNode.keyCount;i++)
					{
						result.push_back(ovwFlowNode.data[i]);

						debug("Data is KEY: %d Ptr: %d %d\n",
								ovwFlowNode.data[i].key,ovwFlowNode.data[i].id.block_num,
								ovwFlowNode.data[i].id.slot_num);
					}
					index = ovwFlowNode.right;

				}while(index);
			}
			else
			{
				result.push_back(Node.data[KIdx - 1]);

				debug("Data is KEY: %d Ptr: %d %d\n",Node.data[KIdx - 1].key,
						Node.data[KIdx - 1].id.block_num,Node.data[KIdx - 1].id.slot_num);
			}


		}
		else
		{
			result.push_back(Node.data[KIdx - 1]);
			debug("Data is KEY: %d Ptr: %d %d\n",Node.data[KIdx - 1].key,
					Node.data[KIdx - 1].id.block_num,Node.data[KIdx - 1].id.slot_num);
		}

	}
	else
	{
		debug("The Key is not found\n");
	}

	return result;

}

void bptree :: bpPrintNode(struct index_entry *Node)
{
	struct data_entry *pData;
	UINT32 Keys[ORDER-1],j;
	UINT32 Pointer[ORDER];

	if(Node->LeafNode)
		{
			printf("Leaf: \n");
			/* Print the Keys */
			pData = (struct data_entry *)Node;

			printf("\t Keys: \t\t");
			for(int k=0; k<pData->keyCount;k++)
				printf("%d \t\t",pData->data[k].key);
			printf("\n");

			/* Print the Pointers */
			printf("\t Pointers: \t");
			for(int l= 0; l<pData->keyCount;l++)
				printf("(%d,%d)\t\t",pData->data[l].id.block_num, pData->data[l].id.slot_num);

			printf("\n\n\n");			
		}
		else
		{
			for(j =0; j<Node->keyCount;j++)
			{
				/* Store the Keys*/
				Keys[j] = Node->key[j];
				Pointer[j] = Node->child[j];
			}
			if(Node->child[j])
				Pointer[j] = Node->child[j];

			/* Print the Keys */

			printf("\t\t Keys: \t\t");
			for(int k=0; k<Node->keyCount;k++)
				printf("%d \t",Keys[k]);
			printf("\n");

			/* Print the Pointers */
			printf("\t\t pointers: \t");
			for(int l=(Node->LeafNode)?1:0; l<Node->keyCount+1;l++)
				printf("%d \t",Pointer[l]);
			printf("\n\n\n");
		}

}

void bptree :: bpPrint()
{
	struct index_entry Node;
	
	for(UINT32 i = 1; i<=header.numBlocks; i++)
	{
		read_block(indexName,i,(BYTE *)&Node);

		printf("\nBlock  %d: \t",i);

		if(i == 1)
			printf("root \n");
		else if(!Node.LeafNode)
			printf("Internal Node \n");

		/* Call the Internal Print Function to print the Node */
		bpPrintNode(&Node);

		}
}

int bptree :: redistribute(struct data_entry *Node, UINT32 KIdx, struct index_data data)
{
	UINT32 k,j;

	if(Node->left)
	{
//		printf("Left Node Search: data: %d \n",data.key);

		struct data_entry leftNode;
		struct index_entry *parent;
		int median = Node->data[0].key;

		read_block(indexName,Node->left,(BYTE *)&leftNode);

		/*If the left Node is not full */
		if(leftNode.keyCount < NUM_KEYS)
		{
			/* Merge these two nodes and redistribute the keys*/
			for(k= leftNode.keyCount,j=0;(k < NUM_KEYS && j < KIdx ); k++,j++)
			{
				leftNode.data[k] = Node->data[j];

				leftNode.keyCount ++;
				Node->keyCount --;
			}
			/*Possible that the key inserted should be in the first key */
			if(data.key < Node->data[j].key)
			{
				if(leftNode.keyCount < NUM_KEYS)
				{
						/* Still there is space */
						leftNode.data[k] = data;
						leftNode.keyCount ++;
				
						if(leftNode.keyCount < NUM_KEYS)
						{
							/*Even Now, there is space in Left Sibing */
							for(;k < NUM_KEYS ; k++,j++)
							{
								leftNode.data[k] = Node->data[j];
								
								leftNode.keyCount ++;
								Node->keyCount --;
							}
						}
						k =0;
					}
				
					else
					{
						/* Insert at the beginning */
						Node->data[0] = data;
						Node->keyCount ++;
						k = 1;
					}

					for(;j<NUM_KEYS;k++,j++)
				{
 					Node->data[k] = Node->data[j]; 
				}

				}
				
				else
					{
						for(k =0;(j<NUM_KEYS && k<Node->keyCount) && (data.key > Node->data[j].key);k++,j++)
				{
 					Node->data[k] = Node->data[j]; 
				}
	if(data.key > Node->data[k-1].key)
				{
					/* Insert at the end */
					Node->data[k] = data;
					Node->keyCount ++;
				}
					}

		/*Update the above layer parents with the new median*/
		for(int n = cacheId - 2; n>=0; n--)
		{
		  parent = (struct index_entry *)&cache[n].cacheBlock;
		  /*Find the place where the median is stored for this child */
		  /* in the parent */
		  for(k=0;k<parent->keyCount;k++)
			if(parent->key[k] == median)
			  break;
		  if(k<parent->keyCount)
		  {
			parent->key[k] = Node->data[0].key;
			/* Write the parent Block */
			write_block(indexName,cache[n].blockNum,(BYTE *)parent);
		  }
		}
		/* Write the Left Node Block */
		write_block(indexName,Node->left, (BYTE *)&leftNode);

		return 1;
	  }		
	}
	else if(Node->right)
	{
		printf("Right Node Search: \n");
	  struct data_entry rightNode;
	  struct index_entry *parent;
	  int median;


	  read_block(indexName,Node->right,(BYTE *)&rightNode);

	  median = rightNode.data[0].key;

	  /*If the left Node is not full */
	  if(rightNode.keyCount < NUM_KEYS)
	  {
		/* Merge these two nodes and redistribute the keys*/
		/*The Key to be inserted must be less than the right Node's first key */
		/* So, just redistribute the Right Node and update the parent node */

		for(k= rightNode.keyCount;k > 1; k--)
		  rightNode.data[k] = rightNode.data[k-1];

		/*Insert the key at the beginning */
		rightNode.data[0] = data;
		rightNode.keyCount ++;

		/*Nothing to be inserted on the Current Node */

		/*Update the upper level parents with the new median*/
		for(int n = cacheId - 2; n>0; n--)
		{
		  parent = (struct index_entry *)cache[n].cacheBlock;

		  /*Find the place where the median is stored for this child */
		  /* in the parent */
		  for(k=0;k<parent->keyCount;k++)
			if(parent->key[k] == median)
			  break;
		  if(k<parent->keyCount)
		  {
			parent->key[k] = rightNode.data[0].key;
			/* Write the parent Block */
			write_block(indexName,cache[n].blockNum, (BYTE *)parent);
		  }
		}

		/* Write the Right Node Block */
		write_block(indexName,Node->right, (BYTE *)&rightNode);

		return 1;
	  }		
	}
	/* Possible case, that both the left and right siblings are full */
	/* In this case, act in the normal way of splitting the node */
	return 0;

}

int bptree ::  InsertKeyOnLeaf(struct data_entry *Node, UINT32 KIdx,
		struct index_data data,UINT32 *Key, UINT32 *Ptr,UINT32 *blockNum) 
{
	struct index_data pData[NUM_KEYS + 1];
	int Count, Count1, Count2 = 0, k,retVal = 0;

	Count = Node->keyCount + 1;
	Count1 = Count < ORDER ? Count : ORDER/2; 
	Count2 = Count - Count1;

	/*This is a special handling for redistribution of Keys
	  within left and right sibling, in case of Node Full */
	if(Count2 && Node->LeafNode)
	{
		/*Redistribute the Key */
		retVal = redistribute(Node,KIdx,data);
		if(retVal)
			return 0;
	}

	for(k = ORDER/2; k < KIdx; k++) 
		pData[k] = Node->data[k];

	pData[KIdx] = data; 

	for(k = KIdx; k < Node->keyCount; k++) 
		pData[k+1] = Node->data[k]; 

	for(k = KIdx; k < Count1; k++) 
		Node->data[k] = pData[k];

	Node->keyCount = Count1;

	if(Count2) 
	{ 
		UINT32 newBlockNum;
		int s, d; 
		struct data_entry NNode; 

		memset(&NNode,0x00,BLKSIZE);

		NNode.LeafNode = 1; 

		for(s = ORDER/2,d = 0; d < Count2; s++, d++)
		{ 
			NNode.data[d] = pData[s]; 
		}
		NNode.keyCount = Count2; 

		//Update the Count of Block Number of the new node
		newBlockNum = ++header.numBlocks;

		*Key = pData[ORDER/2].key;
		*Ptr = newBlockNum; 

		if(*blockNum == 1)
		{
			//Create the root node
			struct index_entry parent;
			memset(&parent,0x00,BLKSIZE);

			parent.keyCount = 1;

			//Update the Node Num
			*blockNum = ++header.numBlocks;

			parent.key[0] = *Key;
			parent.child[0] = *blockNum;
			parent.child[1] = *Ptr;

			header.numLevels++;

			write_block(indexName,1, (BYTE *)&parent);

			Count2=0;
		}


		//Update left Sibling Pointer
		NNode.left = *blockNum;

		NNode.right = Node->right;
		Node->right = newBlockNum; 
		//Update Head
		write_block(indexName,0, (BYTE *)&header);

		//Update the new Node
		write_block(indexName,newBlockNum, (BYTE *)&NNode);

	}

	return Count2;
}

int bptree :: InsertKeyOnNonLeaf(struct index_entry *Node, UINT32 KIdx, 
		UINT32 *Key, UINT32 *Ptr,UINT32 *blockNum) 
{ 
	long Keys[ORDER], Ptrs[ORDER+1];
	int Count, Count1, Count2, k; 
	Count = Node->keyCount + 1;
	Count1 = Count < ORDER ? Count : ORDER/2; 
	Count2 = Count - Count1;
	int newBlockNum;

	for(k = ORDER/2; k < KIdx; k++) 
	{ 
		Keys[k] = Node->key[k];
		Ptrs[k+1] = Node->child[k+1]; 
	} 

	Keys[KIdx] = *Key; 
	Ptrs[KIdx+1] = *Ptr;
	for(k = KIdx; k < Node->keyCount; k++) 
	{ 
		Keys[k+1] = Node->key[k]; 
		Ptrs[k+2] = Node->child[k+1]; 
	}
	for(k = KIdx; k < Count1; k++) 
	{ 
		Node->key[k] = Keys[k];
		Node->child[k+1] = Ptrs[k+1]; 
	}
	Node->keyCount = Count1;

	if(Count2) 
	{ 
		int s, d; 
		struct index_entry NNode; 

		memset(&NNode,0x00,BLKSIZE);

		NNode.LeafNode = Node->LeafNode; 
		Count2 -= !Node->LeafNode; 
		for(s = ORDER/2 + !Node->LeafNode, d = 0; d < Count2; s++, d++)
		{ 
			NNode.key[d] = Keys[s]; 
			NNode.child[d] = Ptrs[s];
		}
		NNode.child[d] = Ptrs[s];
		NNode.keyCount = Count2; 

		//Update the Block Number of the new node
		newBlockNum = ++header.numBlocks;

		*Key = Keys[ORDER/2];
		*Ptr = header.numBlocks; 


		if(*blockNum == 1)
		{
			//Create the root node
			struct index_entry parent;

			parent.left = NULL;
			parent.right = NULL;
			parent.LeafNode = 0;
			parent.keyCount = 1;

			//Update the Node Num
			*blockNum = ++header.numBlocks;


			parent.key[0] = *Key;
			parent.child[0] = *blockNum;
			parent.child[1] = newBlockNum;

			header.numLevels++;

			write_block(indexName,1,(BYTE *)&parent);

			Count2=0;
		}

		//Update left Sibling Pointer
		NNode.left = *blockNum;

		NNode.right = Node->right;
		Node->right = newBlockNum; 

		//Update Head
		write_block(indexName,0,(BYTE *)&header);

		//Update the new Node
		write_block(indexName,newBlockNum,(BYTE *)&NNode);
	} 

	return Count2;
}

void bptree :: HandleDup(struct data_entry *Node, struct index_data &data,
		UINT32 keyIndex,UINT32 blockNum)
{

	/* Check the Child[0] to see if there is a overflow bucket that exists*/
	if(checkIfOvwFlowExists(Node,keyIndex))	{
		/*OverFlow Bucket already exists */
		struct data_entry ovwNode;
		read_block(indexName,Node->data[keyIndex].id.block_num,(BYTE *)&ovwNode);

		/*Check if it is full */
		if(ovwNode.keyCount < NUM_KEYS)
		{
			/* It is not full */
			/* Insert the data here */
			ovwNode.data[ovwNode.keyCount] = data;

			ovwNode.keyCount ++;

			/* Write the Node & Exist */
			write_block(indexName,Node->data[keyIndex].id.block_num,
			(BYTE *)&ovwNode);

		}
		else
		{
			/*Create a New one */
			struct data_entry newOvwNode;
			UINT32 newBlockNum,oldBlockNum;

			newBlockNum = ++header.numBlocks;
			oldBlockNum = Node->data[keyIndex].id.block_num;

			memset(&newOvwNode,0x00,BLKSIZE);

			newOvwNode.data[0] = data;
			newOvwNode.keyCount++;

			newOvwNode.LeafNode = 1;

			/*Update the Pointers */

			/*Insert each OverFlow node to the left of the existing OverFlow Node */

			newOvwNode.right = oldBlockNum;
			newOvwNode.left = 0;
			
			/*Update the Existing OverFlowBlock */
			ovwNode.left = newBlockNum;
			Node->data[keyIndex].id.block_num = newBlockNum;

			/*Update the LeafNode as well */
			/*To point to the new overflow node */
			/*Write the New OverFlowBlock */
			write_block(indexName,newBlockNum,(BYTE *)&newOvwNode);

			/* Write the Node */
			write_block(indexName,oldBlockNum,(BYTE *)&ovwNode);

			/*Update the Leaf Block */
			write_block(indexName,blockNum,(BYTE *)Node);

			/*Write the Header */
			write_block(indexName,0,(BYTE *)&header);
		}

	}
	else
	{
		/*Create a OverFlow Bucket for this key*/

		struct data_entry newOvwNode;
		memset(&newOvwNode,0x00,BLKSIZE);

		/*First, copy the data from Node to the OverFlow Node*/
		newOvwNode.data[0] = Node->data[keyIndex];
		newOvwNode.data[1] = data;

		newOvwNode.keyCount+=2;
		newOvwNode.left = 0;
		newOvwNode.right = 0;

		newOvwNode.LeafNode = 1;

		/*Write the New Block */
		write_block(indexName,++header.numBlocks,(BYTE *)&newOvwNode);

		/*Write the Header */
		write_block(indexName,0,(BYTE *)&header);

		/*Update the Mask in Leaf Node */
		switch(keyIndex)
		{
			case 0: Node->ovwFlowMask|= MASK_FIRST_KEY_OVERFLOW ; break;
			case 1: Node->ovwFlowMask|= MASK_SECOND_KEY_OVERFLOW; break;
			case 2: Node->ovwFlowMask|= MASK_THIRD_KEY_OVERFLOW; break;
			case 3: Node->ovwFlowMask|= MASK_THIRD_KEY_OVERFLOW; break;
		}

		/*Update the Pointer */
		Node->data[keyIndex].id.block_num = header.numBlocks;
		Node->data[keyIndex].id.slot_num = 0;
		/*Update the Block */
		write_block(indexName,blockNum,(BYTE *)Node);

	}
}

int bptree :: RecInsert(struct index_data data) 
{ 
	struct data_entry Node;
	struct index_entry *pNode;

	UINT32 Key = data.key;
	UINT32 Ptr = 0;

	UINT32 KIdx, Split = 0;
	UINT32 EqualKey = 0; 

	UINT32 blockNum = 0;

	pNode = (struct index_entry *)&Node;

	//Lookup and get the correct Node
	Lookup(data.key,pNode,0);
	
	if(cacheId >= 1 )
	{
		blockNum = cache[cacheId-1].blockNum;
//		printf("Block Num: %d\n",blockNum);
	}
	KIdx = FindKey(&Node, Key);

	//Have got the correct Node where it has to be inserted
	EqualKey = KIdx && Node.data[KIdx-1].key == Key;

	if(EqualKey)
	{
		if(header.dupAllowed)
		{
	//		printf("Calling HandleDup Func: \n");
			/* (KIdx - 1) is the location where the Key has been found */
			HandleDup(&Node,data,KIdx-1,blockNum);
	//		printf("Returing \n");

			return false;
		}
		else
		{
			printf(" No duplicates allowed : %d \n",Key);
			return true;
		}
	}

	/*New Key insert normally */
	do
	{
	//		printf("Doing a Normal Insert: \n");

		if(Node.LeafNode)
		{
			Split = InsertKeyOnLeaf(&Node, KIdx, data, &Key, &Ptr, &blockNum); 
			//write to file
			write_block(indexName, blockNum,(BYTE *) &Node);
		}
		else
		{
			Split = InsertKeyOnNonLeaf(pNode, KIdx, &Key, &Ptr,&blockNum); 
			//write to file
			write_block(indexName,blockNum,(BYTE *)pNode);
		}

		//The top most Node would be the Leaf Node.
		if(Node.LeafNode && cacheId)
			cacheId--;

		//Load the parent
		if(cacheId >0 && Split)
		{
			memcpy(pNode,&cache[--cacheId].cacheBlock,BLKSIZE);

			blockNum = cache[cacheId].blockNum;

			KIdx = FindKey(&Node, Key);
		}

	}while(Split);

	cacheId = 0;
	return false; 
} 


void bptree::read_block(string &tblname, UINT32 block_num, BYTE *data)
{
	bm.read_file(tblname, block_num, data);
}

void bptree::write_block(string &tblname, UINT32 block_num, BYTE *data)
{
	bm.write_file(tblname, block_num, data);
}


